package cn.cutie.clot.mq.client;

import cn.cutie.clot.mq.model.ClotMessage;
import lombok.Getter;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Description:
 * @Author: Cutie
 * @CreateDate: 2024/6/27 12:36
 * @Version: 0.0.1
 */
public class ClotConsumer<T> {

    private String id;
    ClotBroker broker;
//    String topic;

//    ClotMq clotMq;

    static AtomicInteger idgen = new AtomicInteger(0);

    public ClotConsumer(ClotBroker broker){
        this.broker = broker;
        this.id = "CID" + idgen.getAndIncrement();
    }

    public void sub(String topic){
        broker.sub(topic, id);
    }

    public void unsub(String topic){
        broker.unsub(topic, id);
    }

    public ClotMessage<T> recv(String topic) {
        return broker.recv(topic, id);
    }

    public boolean ack(String topic, int offset) {
        return broker.ack(topic, id, offset);
    }

    public boolean ack(String topic, ClotMessage<?> message) {
        int offset = Integer.parseInt(message.getHeaders().get("X-offset"));
        return ack(topic, offset);
    }

    public void listen(String topic, ClotListener<T> listener){
        this.listener = listener;
        broker.addConsumer(topic, this);
    }

    @Getter
    private ClotListener listener;
}
